package util;

import data.DataPoint;
import fliter.SensorIdFliter;
import org.apache.beam.sdk.transforms.DoFn;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by emily on 17-12-12.
 */
public class PositiveMap extends DoFn<DataPoint, DataPoint> {

    private static final Logger LOG = LoggerFactory.getLogger(PositiveMap.class);

    @ProcessElement
    public void processElement(ProcessContext c){

        Double value;
        //LOG.info("dp timestamp:" + c.timestamp());
        if(c.element().value<0){
            //LOG.info("below 0 and map to 0");
            //LOG.info(c.element().value.toString());
            value = 0.0;
        }else{
            value = c.element().value;
        }
        DataPoint dp = new DataPoint();
        dp.setDataPoint(c.element().sensorId, c.element().timestamp, value);
        c.output(dp);


    }
}
